package com.mutouren.modules.datapush.local;

import java.io.IOException;
import java.util.Date;

import org.apache.commons.lang3.StringUtils;

import com.alibaba.fastjson.JSON;
import com.mutouren.common.mq.kafka.Producer;
import com.mutouren.modules.datapush.base.Constant;
import com.mutouren.modules.datapush.base.PushMessage;
import com.mutouren.modules.datapush.base.PushStatus;
import com.mutouren.modules.datapush.server.AbstractServer;
import com.mutouren.modules.datapush.strategy.ConcurrentStrategy;
import com.mutouren.modules.datapush.strategy.RepeatStrategy;

public class KafkaServerProxy extends AbstractServer{
	
	private final static String proxyMethodAlias = Constant.defaultMethodAlias;
	private Producer<Integer, byte[]> kafkaProducer;
	private final static String topic = "mtr-datapush";

	public KafkaServerProxy(String serverAlias, String bootstrapServers) {
		super(serverAlias, 1);
		if (StringUtils.isBlank(bootstrapServers)) {
			throw new RuntimeException("bootstrapServers is blank");
		}				
		kafkaProducer = new Producer<Integer, byte[]>(bootstrapServers, byte[].class);
		init();
	}

	private void init() {
		repeatStrategyManager.addRepeatStrategy(new RepeatStrategy(
				proxyMethodAlias, 2, 8, 
				20, 2, "ok"));		
	}	

	@Override
	protected void send(PushMessage message) {
		message.setRequestCount(message.getRequestCount() + 1);
		message.setLastUpdateTime(new Date());
		
		try{
			kafkaProducer.send(topic, message.encode());
			message.setPushStatus(PushStatus.SUCCESS);
			message.setResponse("ok");
		} catch (Throwable t) {
			message.setPushStatus(PushStatus.EXCEPTION);
			message.setResponse(StringUtils.substring(t.getMessage(), 0, 100));			
			errorLogger.error(this.getServerAlias() + " send pushMessage error", t);
		} finally {		
			sendLogger.debug(String.format("serverAlias:%s, message:%s", this.getServerAlias(), JSON.toJSONString(message)));
		}		
	}
	
	@Override
	public void close() {		
		try {
			kafkaProducer.close();
		} catch (IOException e) {
			errorLogger.error(this.getServerAlias() + " kafkaProducer close error", e);
		}
		super.close();
	}

	@Override	
	protected boolean isExpired(PushMessage message) {
		return repeatStrategyManager.isExpired(proxyMethodAlias, message);
	}
	@Override		
	protected boolean isPermitSend(PushMessage message) {
		return repeatStrategyManager.isPermitSend(proxyMethodAlias, message);
	}
	@Override		
	protected boolean checkSuccessFlag(String methodAlias, String value) {
		return repeatStrategyManager.checkSuccessFlag(proxyMethodAlias, value);
	}
	@Override		
	protected boolean isPermitRepeat(PushMessage message) {
		return repeatStrategyManager.isPermitRepeat(proxyMethodAlias, message);
	}
	
	@Override
	public void addConcurrentStrategy(ConcurrentStrategy strategy) {
		throw new IllegalArgumentException("in proxy mode, not permit add concurrentStrategy");
	}	
	@Override
	public void addRepeatStrategy(RepeatStrategy strategy) {
		throw new IllegalArgumentException("in proxy mode, not permit add repeatStrategy");
	}	

}
